package com.ruoyi.kafka;

import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.config.ConfigResource;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @program: ruoyi
 * @description: KafkaAdminTest
 * @author: zengchen
 * @create: 2020-10-15 16:39
 **/
public class KafkaAdminTest {

    public static final String TOPIC_NAME = "study_test";

    public static final String BOOTSTRAP_SERVERS_CONFIG = "centos7-out:9092";

    private static AdminClient adminClient = null;

    public static void main(String[] args) throws Exception {
        adminClient = createAdminClient();
        // 1.查看topiclist
        listTopics();

//        System.out.println("-----------------------");

        // 2.查看包含内部主题的topiclist
//        listInternalTopics();

        // 3.createTopic
//        createTopic();

        // 4.deleteTopic
//        deleteTopic();

//        System.out.println("-----------------------");

        // 5.描述Topic
//        describeTopics();

        // 6.描述kafka配置 (BROKER_LOGGER, BROKER, TOPIC)
//        describeConfig();

//        System.out.println("-----------------------");

        // 7.修改kafka配置 (BROKER_LOGGER, BROKER, TOPIC)
//        modifyConfig();

        // 8.增加partition数量
        incrPartitions(3);

    }

    // 8.增加partition数量
    private static void incrPartitions(int i) throws ExecutionException, InterruptedException {
        Map<String, NewPartitions> newPartitions = new HashMap<>();
        NewPartitions partitions = NewPartitions.increaseTo(i);
        newPartitions.put(TOPIC_NAME, partitions);
        adminClient.createPartitions(newPartitions);

        describeTopics();
    }

    private static void modifyConfig() throws ExecutionException, InterruptedException {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        ConfigEntry configEntry = new ConfigEntry("preallocate","true");
        AlterConfigOp alterConfigOp = new AlterConfigOp(configEntry,AlterConfigOp.OpType.SET);
        Map<ConfigResource,Collection<AlterConfigOp>> configMaps = new HashMap<>();
        configMaps.put(configResource,Arrays.asList(alterConfigOp));
        adminClient.incrementalAlterConfigs(configMaps);

        describeConfig();
    }

    // 6.描述kafka配置 (BROKER_LOGGER, BROKER, TOPIC)
    private static void describeConfig() throws ExecutionException, InterruptedException {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource));
        Map<ConfigResource, KafkaFuture<Config>> values = describeConfigsResult.values();
        KafkaFuture<Config> configKafkaFuture = values.get(configResource);
        Config config = configKafkaFuture.get();
        Collection<ConfigEntry> entries = config.entries();
        entries.forEach(e->{
            System.out.println(ToStringBuilder.reflectionToString(e));
        });
    }

    // 5.描述Topic
    private static void describeTopics() throws ExecutionException, InterruptedException {
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME));
        Map<String, KafkaFuture<TopicDescription>> values = describeTopicsResult.values();
        KafkaFuture<TopicDescription> topicDescriptionKafkaFuture = values.get(TOPIC_NAME);
        TopicDescription topicDescription = topicDescriptionKafkaFuture.get();
        System.out.println(ToStringBuilder.reflectionToString(topicDescription));

    }

    private static void deleteTopic() throws Exception {
        adminClient.deleteTopics(Arrays.asList(TOPIC_NAME));
        listTopics();
    }

    // 查看包含内部主题的topiclist
    private static void listInternalTopics() throws ExecutionException, InterruptedException {
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        listTopicsOptions.listInternal(true);
        ListTopicsResult listTopicsResult = adminClient.listTopics(listTopicsOptions);
        KafkaFuture<Collection<TopicListing>> listings = listTopicsResult.listings();
        Collection<TopicListing> topicListings = listings.get();
        topicListings.forEach(e -> {
            System.out.println("listInternalTopics: name=" + e.name() + ",isInternal=" + e.isInternal());
        });
    }

    // createTopic
    private static void createTopic() throws Exception {
        adminClient.createTopics(Arrays.asList(new NewTopic(TOPIC_NAME, 1, Short.parseShort("1"))));
        listTopics();
    }

    // 查看topiclist
    private static void listTopics() throws Exception {
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        KafkaFuture<Collection<TopicListing>> listings = listTopicsResult.listings();
        Collection<TopicListing> topicListings = listings.get();
        topicListings.forEach(e -> {
            System.out.println("listTopics: name=" + e.name() + ",isInternal=" + e.isInternal());
        });
    }

    private static AdminClient createAdminClient() {
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "centos7-out:9092");
        return KafkaAdminClient.create(properties);
    }

}
